#!bin/bash

if [ -n "$1" ] ; then
  data_date=$1
else
  data_date=`date -d '-1 day' +%F`
fi

TMP_EVENT_LOG_CLEANED_SQL="
DROP TABLE IF EXISTS car_home_warehouse.tmp_event_log_cleaned;
CREATE TABLE IF NOT EXISTS car_home_warehouse.tmp_event_log_cleaned
AS
SELECT log_id,
       if(user_id = '', 'unknown', user_id)   as user_id,
       session_id,
       device_id,
       search_keyword,
       unix_timestamp(search_time)            as search_time,
       COALESCE(province, 'unknown_province') AS province,
       COALESCE(city, 'unknown_city')         AS city
FROM car_home_warehouse.auto_home_search_log
WHERE dt = '${data_date}'
  and device_id IS NOT NULL
  and device_id != ''
    LIMIT 100000
;
"

TMP_EVENT_LOG_SESSION_SQL="
DROP TABLE IF EXISTS car_home_warehouse.tmp_event_log_session;
CREATE TABLE car_home_warehouse.tmp_event_log_session
AS
select log_id,
       user_id,
       device_id,
       search_keyword,
       search_time,
       province,
       city,
       last_value(new_session) over (partition by session_id order by search_time) as new_session
from (select log_id,
             user_id,
             session_id,
             device_id,
             search_keyword,
             search_time,
             province,
             city,
             if(diff_dt / (1000 * 60 * 10) >= 10, concat(session_id, '_', search_time), '') as new_session
      from (select log_id,
                   user_id,
                   session_id,
                   device_id,
                   search_keyword,
                   search_time,
                   province,
                   city,
                   (search_time -
                    lag(`search_time`, 1, 0L) over (partition by session_id order by search_time)) as diff_dt
            from car_home_warehouse.tmp_event_log_cleaned
           ) a1) b2;
"

TMP_EVENT_LOG_KEY_SQL="
DROP TABLE IF EXISTS car_home_warehouse.tmp_event_log_key;
CREATE TABLE car_home_warehouse.tmp_event_log_key
AS
SELECT tmp.log_id,
       tmp.user_id,
       tmp.new_session,
       tmp.device_id,
       dim.brand_id,
       tmp.search_keyword,
       dim.series_id,
       dim.model_id,
       dim.price_range,
       from_unixtime(tmp.search_time,'yyyy-MM-dd hh:mm:ss') as search_time ,
       tmp.province,
       tmp.city
FROM car_home_warehouse.tmp_event_log_session tmp
         LEFT JOIN
     car_home_warehouse.dim_brand_series_model_key_word dim
     on dim.dt = '${data_date}' and dim.normalized_keyword = keyword
    LIMIT 50000;
"

DWD_AUTO_HOME_SEARCH_LOG_SQL="
INSERT OVERWRITE TABLE car_home_warehouse.dwd_auto_home_search_log partition (dt = '${data_date}')
select log_id,
       user_id,
       new_session,
       device_id,
       brand_id,
       search_keyword,
       series_id,
       model_id,
       price_range,
       search_time,
       province,
       city
from car_home_warehouse.tmp_event_log_key;
"

/opt/module/spark/bin/beeline -u jdbc:hive2://node101:10001 -n bwie -p 123456 -e "${TMP_EVENT_LOG_CLEANED_SQL}${TMP_EVENT_LOG_SESSION_SQL}${TMP_EVENT_LOG_KEY_SQL}${DWD_AUTO_HOME_SEARCH_LOG_SQL}"

